package org.example.server;


import cn.hutool.json.JSONUtil;
import com.google.common.collect.Lists;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.logging.ByteBufFormat;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.NetServer;
import io.vertx.core.net.NetServerOptions;
import io.vertx.mqtt.MqttClient;
import org.example.domain.BuyOrSell;
import org.example.domain.Sell;
import org.example.domain.Three;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.CrossOrigin;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.stream.Collectors;
@CrossOrigin
@Component
@RequestMapping("/sell")
@RestController
public class TCPServer {

    public static final BlockingQueue<Sell> queue = new LinkedBlockingQueue<>();
    public static final List<Sell> msgList = Lists.newArrayList();
    private static final int tcp_port = 8891;

    @RequestMapping("/sellFive")
    public List<BuyOrSell> sellFive(Integer stockId){
        int i=0;
        List<Sell> buyFiveForQueue = getSellFiveByMsgList(queue,stockId);
        List<BuyOrSell> buyList = new ArrayList<>();
        for (Sell sell : buyFiveForQueue) {
            i++;
            BuyOrSell buyOrSell = new BuyOrSell();
            buyOrSell.setName("卖"+i);
            buyOrSell.setSellPrice(sell.getSellPrice());
            buyOrSell.setBuyNum(sell.getNum());
            buyOrSell.setType("sell");
            buyList.add(buyOrSell);
        }
        return buyList;
    }

    private List<Sell> getSellFiveByMsgList(BlockingQueue<Sell> queue, Integer stockId) {
        List<Sell> filterByStockId = msgList.stream()
                .filter(item -> item.getStockId() == stockId)
                .collect(Collectors.toList());
        // 按照价格对筛选后的列表进行降序排序
        filterByStockId.sort(Comparator.comparing(Sell::getSellPrice));
        List<Sell> top5ByStockId = filterByStockId.subList(0, Math.min(5, filterByStockId.size()));
        return top5ByStockId;
    }

    @PostConstruct
    public void init() {
        Vertx vertx = Vertx.vertx();
        NetServerOptions options = new NetServerOptions()
                .setLogActivity(true)
                .setActivityLogDataFormat(ByteBufFormat.SIMPLE);
        NetServer server = vertx.createNetServer(options);
        server.connectHandler(socket -> {
            socket.handler(buffer -> {
//                System.out.println("Received data: " + buffer.toString());
                int anInt = buffer.getInt(0);
                System.err.println(anInt);
                String string = buffer.getString(4, 4 + anInt);
//                String string = buffer.toString();
                Sell sell = JSONUtil.toBean(string, Sell.class);

                queue.offer(sell);
                try {

                    msgList.add(queue.take());
                    System.out.println(msgList);
                    List<Sell> sells = ListSort();
                    System.out.println(sells);
                    MQTTClient(sells);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }

            });


        });
        server.listen(tcp_port, "127.0.0.1", res -> {
            if (res.succeeded()) {
                System.out.println("监听成功");
            } else {
                System.out.println("监听失败");
            }
        });
    }

    public List<Sell> ListSort() {
        msgList.sort(
                Comparator.comparingDouble(Sell::getSellPrice)
                        .thenComparingLong(Sell::getCreateDate));
        return msgList;
    }

    public void MQTTClient(List<Sell> sells) {
        Vertx vertx = Vertx.vertx();
        String s = JSONUtil.toJsonStr(sells);
        MqttClient mqttClient = MqttClient.create(vertx);
        mqttClient.connect(1883, "127.0.0.1", res -> {
            if (res.succeeded()) {
                String topic = "MyTopic";
                mqttClient.publish(topic, Buffer.buffer().appendString(s), MqttQoS.AT_MOST_ONCE, false, false);

            } else {
                System.out.println("连接失败");
            }
        }).exceptionHandler(throwable -> System.out.println("连接mqtt异常"));
    }
}
